package push

import (
	npb "enotify/api/notify"
	"enotify/pkg/async/conf"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/golang/protobuf/proto"
	"github.com/golang/protobuf/ptypes"
	"github.com/golang/protobuf/ptypes/any"
	kafka "gopkg.in/Shopify/sarama.v1"
	"strconv"
)


type Group struct {
	 Id int
	 SceneId int64
	 CsvDir string
}

type Push struct {
	kafkaPub kafka.SyncProducer
}

func newKafkaPub(c *conf.Kafka) kafka.SyncProducer {
	kc := kafka.NewConfig()
	kc.Producer.RequiredAcks = kafka.WaitForAll // Wait for all in-sync replicas to ack the message
	kc.Producer.Retry.Max = 10                  // Retry up to 10 times to produce the message
	kc.Producer.Return.Successes = true
	pub, err := kafka.NewSyncProducer(c.Brokers, kc)
	if err != nil {
		panic(err)
	}
	return pub
}
func NewPush(c *conf.Config) *Push {
	return &Push{
		kafkaPub: newKafkaPub(c.Kafka),
	}
}

func (n *Push) Stop() {
	n.kafkaPub.Close()
}
func (n *Push) Push(reqId int64, bizType int32, sceneId int64, data *any.Any) (err error) {
	var info npb.SendMsg
	err = ptypes.UnmarshalAny(data, &info)
	if err != nil {
		return
	}
	n.pushMsg(reqId, bizType, sceneId, &info)
	return
}
func (n *Push) pushMsg(reqId int64, bizType int32, sceneId int64, message *npb.SendMsg) (err error) {
	sendmsg, err := proto.Marshal(message)
	if err != nil {
		return
	}
	m := &kafka.ProducerMessage{
		Topic: "test",
		Key:   sarama.StringEncoder(strconv.FormatInt(reqId, 10)),
		Value: sarama.ByteEncoder(sendmsg),
	}
	if _, _, err = n.kafkaPub.SendMessage(m); err != nil {
		err = fmt.Errorf("PushMsg.send(push pushMsg:%v) error(%v)", message, err)
	}
	return
}
